package SparkGraphXInAction

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.SparkContext._
import org.apache.spark.graphx._
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators

/**
  * Created by Administrator on 2017/5/8 0008.
  * 如果没有checkpoint，迭代500次就会出现栈溢出。
  */
object TestCheckpoint {
  //  def main(args: Array[String]): Unit = {
  //    // 屏蔽日志
  //    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
  //    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
  //
  //    //设置运行环境
  //    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
  //    val sc = new SparkContext(conf)
  //    val iterations =500
  //    var g = Graph.fromEdges(sc.makeRDD(Seq(Edge(1L,3L,1), Edge(2L, 4L, 1), Edge(3L, 4L, 1))),1)
  //    for(i <- 1 to iterations){
  //      println("Iteration: "+ i)
  //      val newGraph: Graph[Int, Int] = g.mapVertices((vid,vd) => (vd*i)/17)
  //      g = g.outerJoinVertices[Int, Int](newGraph.vertices){
  //        (vid, vd, newData) => newData.getOrElse(0)
  //      }
  //    }
  //    g.vertices.collect.foreach(println)
  //  }

  def main(args: Array[String]): Unit = {
    // 屏蔽日志
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    //设置checkpoint目录，如果不设置，那么随后的checkpoint不会有任何效果。
    sc.setCheckpointDir("E:\\spark-checkpoint")
    //记录更新了多少次图。
    var updateCount = 0
    //每50次图更新做一次checkpoint
    val checkpointInterval = 50

    def update(newData: Graph[Int, Int]): Unit = {
      newData.persist()
      updateCount += 1
      if (updateCount % checkpointInterval == 0) {
        newData.checkpoint()
      }
    }

    val iterations = 500
    var g = Graph.fromEdges(sc.makeRDD(Seq(Edge(1L, 3L, 1), Edge(2L, 4L, 1), Edge(3L, 4L, 1))), 1)
    update(g)
    g.vertices.count
    for (i <- 1 to iterations) {
      println("Iteration : " + i)
      val newGraph: Graph[Int, Int] = g.mapVertices((vid, vd) => (vd * i) / 17)
      g = g.outerJoinVertices[Int, Int](newGraph.vertices) {
        (vid, vd, newData) => newData.getOrElse(0)
      }
      update(g)
      g.vertices.count
    }
    g.vertices.collect.foreach(println)
  }
}